package com.small.nacos.core.cluster;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.common.http.Callback;
import com.alibaba.nacos.common.http.HttpClientBeanHolder;
import com.alibaba.nacos.common.http.HttpUtils;
import com.alibaba.nacos.common.http.client.NacosAsyncRestTemplate;
import com.alibaba.nacos.common.http.param.Header;
import com.alibaba.nacos.common.http.param.Query;
import com.alibaba.nacos.common.model.RestResult;
import com.alibaba.nacos.common.utils.ConcurrentHashSet;
import com.alibaba.nacos.common.utils.ExceptionUtil;
import com.alibaba.nacos.common.utils.VersionUtils;
import com.google.common.reflect.TypeToken;
import com.small.nacos.common.notify.Event;
import com.small.nacos.common.notify.NotifyCenter;
import com.small.nacos.core.cluster.lookup.LookupFactory;
import com.small.nacos.core.utils.Commons;
import com.small.nacos.naming.misc.GlobalExecutor;
import com.small.nacos.sys.env.Constants;
import com.small.nacos.sys.env.EnvUtil;
import com.small.nacos.sys.utils.InetUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.web.context.WebServerInitializedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Author zhoujin
 * @Date 2022/10/31 20:00
 */
@Component(value = "serverMemberManager")
public class ServerMemberManager implements ApplicationListener<WebServerInitializedEvent> {

    protected static final Logger LOGGER = LoggerFactory.getLogger(ServerMemberManager.class);

    private volatile ConcurrentSkipListMap<String, Member> serverList;

    private final NacosAsyncRestTemplate asyncRestTemplate = HttpClientBeanHolder
            .getNacosAsyncRestTemplate(LOGGER);

    /**
     * 当前节点端口号
     */
    private int port;

    /**
     * 当前节点请求地址
     */
    private String localAddress;

    /**
     * 当前节点
     */
    private volatile Member self;

    /**
     * 成员查找
     */
    private MemberLookup lookup;

    private final MemberInfoReportTask infoReportTask = new MemberInfoReportTask();

    public ServerMemberManager() throws Exception {
        this.serverList = new ConcurrentSkipListMap();
        init();
    }

    public void init() throws NacosException {
        this.port = EnvUtil.getProperty("server.port", Integer.class, 8848);
        this.localAddress = InetUtils.getSelfIP() + ":" + port;
        this.self = MemberUtil.singleParse(this.localAddress);
        serverList.put(self.getAddress(), self);

        //初始化服务改变事件
        registerClusterEvent();

        initAndStartLookup();

        if (serverList.isEmpty()) {
            throw new NacosException(NacosException.SERVER_ERROR, "cannot get serverlist, so exit.");
        }

    }

    private void initAndStartLookup() throws NacosException {
        this.lookup = LookupFactory.createLookUp(this);
        this.lookup.start();
    }

    private void registerClusterEvent() {
        // Register node change events
        NotifyCenter.registerToPublisher(MembersChangeEvent.class,
                EnvUtil.getProperty("nacos.member-change-event.queue.size", Integer.class, 128));
    }


    /**
     * 合并成员
     *
     * @param members
     * @return
     */
    synchronized boolean memberChange(Collection<Member> members) {
        if (members == null || members.size() == 0) {
            return false;
        }

        boolean isContainSelfIp = members.stream().anyMatch(selfMember -> Objects.equals(localAddress, selfMember.getAddress()));
        if (!isContainSelfIp) {
            members.add(this.self);
        }

        boolean hasChange = members.size() != serverList.size();
        ConcurrentSkipListMap<String, Member> tmpMap = new ConcurrentSkipListMap<>();


        for (Member member : members) {

            final String address = member.getAddress();
            if (!serverList.containsKey(address)) {
                hasChange = true;

                member.setState(NodeState.DOWN);
            } else {
                member.setState(serverList.get(address).getState());
            }
            tmpMap.put(member.getAddress(), member);
        }

        serverList = tmpMap;

        //同步数据到文件中,这里会将原来的serverList跟换
        if (hasChange) {
            MemberUtil.syncToFile(allMembers());
            notifyMemberChange();
        }
        return hasChange;
    }

    public Collection<Member> allMembers() {
        HashSet<Member> set = new HashSet<>(serverList.values());
        set.add(self);
        return set;
    }

    public List<Member> allMembersWithoutSelf() {
        List<Member> members = new ArrayList<>(serverList.values());
        members.remove(self);
        return members;
    }

    public Member getSelf() {
        return this.self;
    }

    //Tomcat启动后会回调ServerMemberManager，开启一个MemberInfoReportTask当前节点信息广播任务。
    @Override
    public void onApplicationEvent(WebServerInitializedEvent event) {
        getSelf().setState(NodeState.UP);
        //不是单应用启动 需要检查Member状态
        if (!EnvUtil.getStandaloneMode()){
            GlobalExecutor.scheduleByCommon(this.infoReportTask, 5_000L);
        }
    }

    void notifyMemberChange() {
        NotifyCenter.publishEvent(MembersChangeEvent.builder().members(allMembers()).build());
    }

    public boolean update(Member newMember) {

        Member member = serverList.get(newMember.getAddress());
        if (member == null){
            return false;
        }
        boolean isPublishChangeEvent = MemberUtil.isBasicInfoChanged(newMember, member);
        MemberUtil.copy(newMember, member);
        if (isPublishChangeEvent){
            notifyMemberChange();
        }
        serverList.put(member.getAddress(),member);
        return true;
    }

    class MemberInfoReportTask extends Task{

        private final TypeToken<RestResult<String>> reference = new TypeToken<RestResult<String>>() {};

        private int cursor = 0;

        @Override
        void executeBody() {
            List<Member> members = allMembersWithoutSelf();
            if (members == null || members.size()==0){
                return;
            }
            cursor = (cursor + 1) % members.size();
            Member target = members.get(cursor);

            final String url = HttpUtils
                    .buildUrl(false, target.getAddress(), EnvUtil.getContextPath(), Commons.NACOS_CORE_CONTEXT,
                            "/cluster/report");

            try {
                Header header = Header.newInstance().addParam(Constants.NACOS_SERVER_HEADER, Constants.VERSION);
                asyncRestTemplate
                        .post(url, header,
                                Query.EMPTY, getSelf(), reference.getType(), new Callback<String>() {

                                    @Override
                                    public void onReceive(RestResult<String> result) {
                                        if (result.getCode() == HttpStatus.NOT_IMPLEMENTED.value()
                                                || result.getCode() == HttpStatus.NOT_FOUND.value()) {
                                            LOGGER.warn("{} version is too low, it is recommended to upgrade the version : {}",
                                                            target, VersionUtils.version);
                                            return;
                                        }
                                        if (result.ok()) {
                                            MemberUtil.onSuccess(ServerMemberManager.this, target);
                                        } else {
                                            LOGGER
                                                    .warn("failed to report new info to target node : {}, result : {}",
                                                            target.getAddress(), result);
                                            MemberUtil.onFail(ServerMemberManager.this, target);
                                        }
                                    }

                                    @Override
                                    public void onError(Throwable throwable) {
                                        LOGGER
                                                .error("failed to report new info to target node : {}, error : {}",
                                                        target.getAddress(),
                                                        ExceptionUtil.getAllExceptionMsg(throwable));
                                        MemberUtil.onFail(ServerMemberManager.this, target);
                                    }

                                    @Override
                                    public void onCancel() {

                                    }
                                });

            } catch (Throwable ex) {
                LOGGER.error("failed to report new info to target node : {}, error : {}", target.getAddress(),
                        ExceptionUtil.getAllExceptionMsg(ex));
            }

        }

        @Override
        protected void after() {
            GlobalExecutor.scheduleByCommon(this, 2000L);
        }
    }

}
